package slh.myMapReduce;

import java.io.IOException;
import java.util.TreeMap;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class FavoriteMovieReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    // 使用TreeMap来存储Top3电影(自动排序)
    private TreeMap<Integer, Text> topMovies = new TreeMap<>();

    @Override
    public void reduce(Text key, Iterable<IntWritable> values, Context context)
            throws IOException, InterruptedException {

        // 计算当前电影的总收藏量
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }

        // 存入TreeMap(使用负数实现降序排序)
        topMovies.put(-sum, new Text(key));

        // 只保留Top3
        if (topMovies.size() > 3) {
            topMovies.remove(topMovies.lastKey());
        }
    }

    @Override
    protected void cleanup(Context context)
            throws IOException, InterruptedException {

        // 输出Top3电影结果
        for (Integer count : topMovies.keySet()) {
            context.write(topMovies.get(count), new IntWritable(-count));
        }
    }
}